package cn.ilikes.tools.nosql.jpa.clients.handers.hbases;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Result;
import org.slf4j.Logger;

import cn.ilikes.tools.logs.Loggerfactory;
import cn.ilikes.tools.nosql.jpa.JPASQLPersistenUnitEnv;
import cn.ilikes.tools.nosql.jpa.clients.dataSources.hbases.HbaseDataSource;
import cn.ilikes.tools.nosql.jpa.clients.handers.AbDataHander;
import cn.ilikes.tools.nosql.jpa.clients.handers.JPASQLDataHnaderException;
import cn.ilikes.tools.nosql.jpa.graphs.Node;
import cn.ilikes.tools.nosql.jpa.metadatas.EntityMetaData;

public class HbaseDataHander extends AbDataHander {

	private Logger logger = Loggerfactory.instance(HbaseDataHander.class);

	private HbaseServices hbaseServices;

	public HbaseDataHander(HbaseDataSource dataSource, JPASQLPersistenUnitEnv persistenUnitEnv) {
		super(dataSource, persistenUnitEnv);
		HbaseServices hbaseServices = new HbaseServices(dataSource, persistenUnitEnv);
		setHbaseServices(hbaseServices);
	}

	private void setHbaseServices(HbaseServices hbaseServices) {
		this.hbaseServices = hbaseServices;
	}

	@Override
	public void save(Node node) throws JPASQLDataHnaderException {
		if (Loggerfactory.isdebug(logger))
			Loggerfactory.debug(logger, "Hbase Data Hander  save : " + node.toString());

		hbaseServices.save(node);
	}

	public void remove(Node node) throws JPASQLDataHnaderException {
		if (Loggerfactory.isdebug(logger))
			Loggerfactory.debug(logger, "Hbase Data Hander  remove  : " + node.toString());

		hbaseServices.remove(node);
	}

	public void flush() {
		if (Loggerfactory.isdebug(logger))
			Loggerfactory.debug(logger, "Hbase Data Hander  flush ");
		hbaseServices.commits();

	}

	@Override
	public Result find(EntityMetaData entityMetaData, Object key) throws JPASQLDataHnaderException {
		if (Loggerfactory.isdebug(logger))
			Loggerfactory.debug(logger, "Hbase Data Hander  find  ");
		return hbaseServices.find(entityMetaData, key);
	}

	@Override
	public byte[] findValue(EntityMetaData entityMetaData, Object key, String family, String qualifier) throws JPASQLDataHnaderException {
		try {
			if (Loggerfactory.isdebug(logger))
				Loggerfactory.debug(logger, "find  Id   : " + key + "  family : " + family + "    qualifier " + qualifier);
			return hbaseServices.getfamilyAndqualifierValues(entityMetaData, key, family, qualifier);
		} catch (IOException e) {
			throw new JPASQLDataHnaderException(e);
		}
	}

	public Result[] seach(EntityMetaData entityMetaData, List keyIds) throws JPASQLDataHnaderException {

		try {
			if (Loggerfactory.isdebug(logger))
				Loggerfactory.debug(logger, "seach  keyIds  : " + keyIds.size());
			String entityTableName = entityMetaData.getEntityTableName();
			HbaseDataSource dataSource = (HbaseDataSource) getDataSource();
			HTablePool tablePool = dataSource.getDataSource();
			HTableInterface hTableInterface = tablePool.getTable(entityTableName);
			List<Get> gets = new ArrayList<Get>();
			for (Object keyId : keyIds) {
				Get get = new Get(HBaseUtils.putToPutByte(keyId));
				gets.add(get);
			}
			Result[] results = hTableInterface.get(gets);
			return results;
		} catch (IOException e) {
			throw new JPASQLDataHnaderException(e);
		}
	}

	@Override
	public Result findId(EntityMetaData entityMetaData, Object key) throws JPASQLDataHnaderException {
		try {
			if (Loggerfactory.isdebug(logger))
				Loggerfactory.debug(logger, "seach  keyIds  : " + key);
			String entityTableName = entityMetaData.getEntityTableName();
			HbaseDataSource dataSource = (HbaseDataSource) getDataSource();
			HTablePool tablePool = dataSource.getDataSource();
			HTableInterface hTableInterface = tablePool.getTable(entityTableName);
			Get get = new Get(HBaseUtils.putToPutByte(key));
			Result results = hTableInterface.get(get);
			return results;
		} catch (IOException e) {
			throw new JPASQLDataHnaderException(e);
		}
	}

	@Override
	public void rollback() {
		if (Loggerfactory.isdebug(logger))
			Loggerfactory.debug(logger, "Hbase  DataHander rollback  ");
	}

}